﻿using Confluent.Kafka;
using Confluent.SchemaRegistry;
using Confluent.SchemaRegistry.Serdes;
using kafkalearn.ProducerTest1.AvroSpecific;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace ProducerTest1
{
    public class AvroSpecificProducer : IPush
    {
        private readonly string _schemaUri;
        private readonly string _topic;
        public AvroSpecificProducer(string schemaUri, string topic)
        {
            _topic = topic;
            _schemaUri = schemaUri;
        }

        public void Invoke(ProducerConfig config)
        {
            throw new NotImplementedException();
        }

        public async Task InvokeAsync(ProducerConfig config)
        {
            SchemaRegistryConfig schemaRegistryConfig = new SchemaRegistryConfig
            {
                Url = _schemaUri
            };
            AvroSerializerConfig avroSerializerConfig = new AvroSerializerConfig
            {
                BufferBytes = 100
            };
            CancellationTokenSource cts = new CancellationTokenSource();
            using(var schemaRegistry = new CachedSchemaRegistryClient(schemaRegistryConfig))
                using(var producer = new ProducerBuilder<string, User>(config)
                .SetKeySerializer(new AvroSerializer<string>(schemaRegistry, avroSerializerConfig))
                .SetValueSerializer(new AvroSerializer<User>(schemaRegistry, avroSerializerConfig))
                .Build())
            {
                Console.WriteLine($"{producer.Name} producing on {_topic}. Enter user names, q to exit.");
                int i = 0;
                string text;
                while((text = Console.ReadLine()) != "q")
                {
                    User user = new User { name = text, favorite_color = "origin", favorite_number = i++, hourly_rate = new Avro.AvroDecimal(67.99) };
                    await producer.ProduceAsync(_topic, new Message<string, User> { Key = text, Value = user })
                        .ContinueWith(task => {
                            if (!task.IsFaulted)
                            {
                                Console.WriteLine($"produced to: {task.Result.TopicPartitionOffset}");
                            }
                            if (task.Exception != null)
                            {
                                Console.WriteLine($"error producing message: {task.Exception.InnerException}");
                            }
                        });
                }
            }
            cts.Cancel();
        }
    }
}
